package com.learn.lb.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author laibo
 * @since 2019/9/3 23:16
 * 从文件系统读取数据
 */
object WordCountFileStreaming {

  def main(args: Array[String]): Unit = {
    //读取文件系统的数据，线程数随意设置
    val sparkConf = new SparkConf().setMaster("local").setAppName("WorkCountFileStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    //注意，这里是监听文件夹下面的文件，当有新的文件被放到设置的文件夹下，则该文件内容会被读取
    //一旦该文件的数据被读取过，再对该文件行进操作（比如追加数据）是不会有效果的
    //move file/cp file 等操作都是可以的。
    val inputDStream = ssc.textFileStream("E:\\idea\\workspace\\spark-learn\\src\\main\\resources")
    val wordCountDStream = inputDStream.flatMap(_.split("_")).map((_, 1)).reduceByKey(_ + _)
    wordCountDStream.print()
    ssc.start()
    ssc.awaitTermination()
  }
}
